package queue

import (
	"context"
	"dealcenter/internal/discovery"
	"dealcenter/internal/model"
	"dealcenter/internal/service"
	"dealcenter/utils/errx"
	"dealcenter/utils/logx"
	"dealcenter/utils/resx"
	"encoding/json"
	"github.com/pkg/errors"
	"github.com/streadway/amqp"
	"strconv"
	"sync"
)

var (
	ce <-chan amqp.Delivery
	cd <-chan amqp.Delivery
)
var wg sync.WaitGroup

func consume(xcg string) {
	var err error

	ce, err = MQChan.Consume(
		pre+RkExec,
		"ctpCashExec",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE] can't get consume"))
	}
	cd, err = MQChan.Consume(
		pre+RkDead,
		"ctpCashDead",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE] can't get consume"))
	}

	var consumers []func(string)
	consumers = append(consumers, exec)
	consumers = append(consumers, dead)

	wg.Add(len(consumers))

	for _, f := range consumers {
		go f(xcg)
	}
}

// =================================== consume logic ===================================

func exec(xcg string) {
	for msg := range ce {
		var err error
		var rl int

		// 获取重试次数
		trl, okt := msg.Headers["retry-left"]
		if !okt {
			rl = 0
		} else {
			var err error
			srl, oks := trl.(string)
			if !oks {
				rl = 0
			} else {
				rl, err = strconv.Atoi(srl)
				if err != nil {
					rl = 0
				}
			}
		}

		// 执行消息
		var data ItemCloseContent
		err = json.Unmarshal(msg.Body, &data)
		if err != nil {
			fail(xcg, msg.Body, rl, err)
			continue
		}
		switch data.ModelType {
		case "sale":
			err = model.UpdateItem(&model.OrderSale{
				SaleId: data.Id,
				State:  data.State,
			})
			if err != nil {
				fail(xcg, msg.Body, rl, err)
				continue
			}
		case "buy":
			err = model.UpdateItem(&model.OrderBuy{
				BuyId: data.Id,
				State: data.State,
			})
			if err != nil {
				fail(xcg, msg.Body, rl, err)
				continue
			}
			if data.State == "closed" {
				buyInfo := model.QueryBuyByBuyId(data.Id)
				cashConn := discovery.CashDiscover.Get()
				if cashConn == nil {
					logx.LogrusObj.Warning("[QUEUE]: Buy order refund failed, reason: ", errx.CtpError(errx.RpcError))
					continue
				}
				cashServiceClient := service.NewCashServiceClient(cashConn)
				rpcResp, err := cashServiceClient.CashAdd(context.Background(), &service.CashChangeRequest{Count: buyInfo.LeftNum * buyInfo.MaxPrice, UserId: buyInfo.BuyerId})
				if rpcResp.GetCode() == 14 {
					discovery.CashDiscover.Sad(cashConn.Target())
				}
				if err != nil || rpcResp.Code != resx.Success {
					logx.LogrusObj.Warning("[QUEUE]: Buy order refund failed, reason: ", resx.GetMsg(int(rpcResp.Code)), err)
					continue
				}
			}
		case "deal":
			err = model.UpdateItem(&model.Deal{
				DealId: data.Id,
				State:  data.State,
			})
			if err != nil {
				fail(xcg, msg.Body, rl, err)
				continue
			}
			if data.State == "closed" {
				dealInfo := model.QueryDealByDealId(data.Id)
				cashConn := discovery.CashDiscover.Get()
				if cashConn == nil {
					logx.LogrusObj.Warning("[QUEUE]: Deal order refund failed, reason: ", errx.CtpError(errx.RpcError))
					continue
				}
				cashServiceClient := service.NewCashServiceClient(cashConn)
				rpcResp, err := cashServiceClient.CashAdd(context.Background(), &service.CashChangeRequest{Count: dealInfo.Num * dealInfo.DealCost, UserId: dealInfo.BuyerId})
				if rpcResp.GetCode() == 14 {
					discovery.CashDiscover.Sad(cashConn.Target())
				}
				if err != nil || rpcResp.Code != resx.Success {
					logx.LogrusObj.Warning("[QUEUE]: Deal order refund failed, reason: ", resx.GetMsg(int(rpcResp.Code)), err)
					continue
				}
			}
		}
	}
}

func dead(xcg string) {
	for msg := range cd {

		// TODO log dead logic
		logx.LogrusObj.Warning("get a dead msg: %s\n", msg.Body)
	}
}

// =================================== republish logic ===================================

func retryMsg(exchange string, msgBody []byte, retryCount string) {
	err := MQChan.Publish(
		exchange,
		RkExec,
		false,
		false,
		amqp.Publishing{
			Headers:         amqp.Table{"retry-left": retryCount},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            msgBody,
		},
	)
	if err != nil {
		panic(err)
	}
}

func killMsg(exchange string, msgBody []byte) {
	err := MQChan.Publish(
		exchange,
		RkDead,
		false,
		false,
		amqp.Publishing{
			Headers:         nil,
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            msgBody,
		},
	)
	if err != nil {
		panic(err)
	}
}

// =================================== publish logic ===================================

// PubItemClose waitType can use queue.RkWait1M / queue.RkWait1H / queue.RkWait1D
func PubItemClose(modelType, waitType string, id int64) error {
	msg, marshalErr := json.Marshal(ItemCloseContent{
		ModelType: modelType,
		Id:        id,
		State:     "closed",
	})
	if marshalErr != nil {
		return marshalErr
	}
	if waitType != RkWait1M && waitType != RkWait1H && waitType != RkWait1D {
		return errors.New("[QUEUE] Bad wait type")
	}
	err := MQChan.Publish(Xcg, waitType, false, false, amqp.Publishing{
		Headers:         amqp.Table{"retry-left": "2"},
		ContentType:     "text/plain",
		ContentEncoding: "",
		Body:            msg,
	})

	return err
}

// =================================================================

func fail(xcg string, msg []byte, left int, reason error) {
	logx.LogrusObj.Errorf("[Queue] Get a exec msg: %s\n, but can't exec: %s", msg, reason.Error())
	if left >= 1 {
		retryMsg(xcg, msg, strconv.Itoa(left-1))
	} else {
		dMsg, _ := json.Marshal(DeadContent{
			Content: msg,
			Reason:  reason.Error(),
		})
		killMsg(xcg, dMsg)
	}
}

type ItemCloseContent struct {
	ModelType string
	Id        int64
	State     string
}
type DeadContent struct {
	Content []byte
	Reason  string
}
